package com.yl.flink.redis;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.io.Closeable;
import java.io.IOException;
import java.util.Objects;

public class RedisContainer implements RedisCmdContainer, Closeable {
    private static final long serialVersionUID = 1L;
    private static final Logger LOG = LoggerFactory.getLogger(RedisContainer.class);
    private final transient JedisPool jedisPool;

    public RedisContainer(JedisPool jedisPool) {
        Objects.requireNonNull(jedisPool, "Jedis Pool can not be null");
        this.jedisPool = jedisPool;
    }

    @Override
    public void close() throws IOException {
        if (this.jedisPool != null) {
            this.jedisPool.close();
        }
    }

    @Override
    public void open() throws Exception {
        this.getInstance().echo("Test");
    }

    @Override
    public void sttl(String key, String value, int ttl) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.set(key, value);
            if (ttl != -1) {
                jedis.expire(key, ttl);
            }
        } catch (Exception var9) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command EXPIRE to key {} with {} error message {}", key, ttl, var9.getMessage());
            }
            throw var9;
        } finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public void rpush(String listName, String value) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.rpush(listName, value);
        } catch (Exception var8) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command RPUSH to list {} error message {}", listName, var8.getMessage());
            }
            throw var8;
        } finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public void lpush(String listName, String value) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.lpush(listName, value);
        } catch (Exception var8) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command LUSH to list {} error message {}", listName, var8.getMessage());
            }
            throw var8;
        } finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public void set(String key, String value) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.set(key, value);
        } catch (Exception var8) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command SET to key {} error message {}", key, var8.getMessage());
            }
            throw var8;
        } finally {
            this.releaseInstance(jedis);
        }
    }

    private Jedis getInstance() {
        return this.jedisPool.getResource();
    }

    private void releaseInstance(Jedis jedis) {
        if (jedis != null) {
            try {
                jedis.close();
            } catch (Exception var3) {
                LOG.error("Failed to close (return) instance to pool", var3);
            }
        }
    }

}
